package com.flinkTest;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.Serializable;
import java.util.Iterator;
import java.util.Random;

/**
 * Created by cola on 2022/5/3.
 */
public class TestSource  implements Iterator<Tuple2<String, Integer>>, Serializable {

    static final String[] NAMES = {"tom", "jerry", "alice", "bob", "john", "grace"};
    static final int GRADE_COUNT = 5;
    static final int SALARY_MAX = 10000;

    private final Random rnd = new Random(hashCode());

    @Override
    public boolean hasNext() {
        return true;
    }

    @Override
    public Tuple2<String, Integer> next() {
        return new Tuple2<>(NAMES[rnd.nextInt(NAMES.length)], rnd.nextInt(GRADE_COUNT) + 1);
    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException();
    }

    public static DataStream<Tuple2<String, Integer>> getSource(
            StreamExecutionEnvironment env, long rate) {
        return env.fromCollection(
                new ThrottledIterator<>(new TestSource(), rate),
                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
    }
}
